#ifndef ROW_AGGREGATOR_H
#define ROW_AGGREGATOR_H

#include "SharedArray.h"
#include "Column.h"
#include "hash.h"

class KeyValue;
class AggregationData;

class RowAggregator {
	public:
		RowAggregator();
		~RowAggregator();

		// Saves configuration, computes key size, value size and default value.
		void init(SharedArray<Column> in, SharedArray<int> groupedColumns, SharedArray<AggregationData> aggregations);

		bool isInitialized() {
			return _defaultValue != 0;
		}

		// Loads input data, computes keys and hashes.
		void loadInputData(SharedArray<Column> in);

		// String of concaternated keys.
		char* concaternatedKeys() const {
			assert(_keys);
			return _keys;
		}

		// Array of hashes.
		hash_t* hashes() const {
			assert(_hashes);
			return _hashes;
		}

		// Size of key in bytes.
		int keySize() const {
			return _keySize;
		}

		// Size of value in bytes.
		int valueSize() const {
			return _valueSize;
		}

		// Amount of loaded data.
		int dataSize() const {
			assert(_in.size() != 0);
			return _in[0].size();
		}

		// Returns default value (base of aggregation
		// equal to zeros for each column type).
		const char* defaultValue() const {
			assert(_defaultValue);
			return _defaultValue;
		}

		// Aggregates loaded input data into given rows.
		void aggregate(SharedArray<char*> valueRows);

		// Loads key-value pairs and converts them to column chunks.
		void loadAggregatedData(SharedArray<KeyValue> keysAndValues, int chunkSize);

		bool hasAggregatedData() const {
			return _aggregatedChunks.size() != 0;
		}

		// Returns next aggregated column chunk.
		SharedArray<Column> takeAggregatedDataChunk();

	private:
		SharedArray<Column> extractToColumns(const SharedArray<KeyValue>& keysAndValues, int startIndex, int amount) const;
		SharedArray<Column> reorderedColumns(const SharedArray<Column>& columns) const;

		RowAggregator(const RowAggregator& rhs);
		RowAggregator& operator=(const RowAggregator& rhs);

		void initKeyValueSizes(const SharedArray<Column>& in);
		void initDefaultValue();
		void computeKeys(const SharedArray<Column>& in);
		void computeHashes(const SharedArray<Column>& in);

		// There are 3 types of column ordering:
		// * input ordering - ordering of columns as in input
		// * internal ordering - used in (key, value) representation;
		//   first keys in order as in aggregations order, then distinct
		//   double values in aggregations order, then distinct int values
		//   in aggregations order, then distinct bool values in aggregations
		//   order
		// * output ordering - first keys then values in aggregations order

		// conversion: input ordering -> internal ordering
		SharedArray<int> _groupedColumns;
		SharedArray<int> _summedIntColumns;
		SharedArray<int> _summedDoubleColumns;
		bool _isCounting;
		// conversion: internal ordering -> output ordering
		SharedArray<int> _outValueColumns;

		SharedArray<Column> _in;
		int _keySize;
		int _valueSize;
		char* _keys;
		hash_t* _hashes;
		char* _defaultValue;
		SharedArray<SharedArray<Column> > _aggregatedChunks;
};

#endif
